package indi.zhifa.study2025.flinkcdc.redis;


import indi.zhifa.study2025.flinkcdc.redis.command.RedisSinkCommand;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.IOException;

@Slf4j
public class SpringDataRedisSink<T> implements Sink<RedisSinkCommand<T>> {

    @Override
    public SinkWriter<RedisSinkCommand<T>> createWriter(InitContext context) throws IOException {
        return null;
    }

    @Override
    public SinkWriter<RedisSinkCommand<T>> createWriter(WriterInitContext context){
        return new LettuceRedisSinkWriter();
    }

    class LettuceRedisSinkWriter implements SinkWriter<RedisSinkCommand<T>> {

        @Override
        public void write(RedisSinkCommand<T> pCmd, Context context) throws IOException, InterruptedException {
            RedisTemplate<String, Object> redisTemplate = RedisConfig.redisTemplate();
            switch (pCmd.getCommand()){
                case SET-> {
                    redisTemplate.opsForValue().set(pCmd.getKey(),pCmd.getValue(),pCmd.getDua());
                }
                case DEL -> {
                    redisTemplate.delete(pCmd.getKey());
                }
            }
        }

        @Override
        public void flush(boolean endOfInput) throws IOException, InterruptedException {

        }

        @Override
        public void close() throws Exception {

        }
    }

}
